package mr.weblog.part1;

import java.io.IOException;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import mr.weblog.support.TextPair;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class WeblogMapper extends Mapper<LongWritable, Text, TextPair, TextPair> {
	private static Logger log = LoggerFactory.getLogger(WeblogMapper.class);

	public final static String htmlGetStr = "GET /";
	public final static String htmlExtStr = ".html";
	public final static String quoteSymbol = "\"";

	private final Text indicator = new Text("L");
	private final Text joinKey = new Text();
	private final Text joinVal = new Text();

	private static Pattern pattern = 
			Pattern.compile("([^ ]*) ([^ ]*) \\[([^]]*)\\] \"([^\"]*)\" ([^ ]*) ([^ ]*)");
	
	private final TextPair joinKeyPair = new TextPair();
	private final TextPair valuePair = new TextPair();
	{
		joinKeyPair.setSide(indicator);
		valuePair.setSide(indicator);
	}

	@Override
	public void map(LongWritable key, Text value, Context context)
			throws IOException, InterruptedException {
		String valStr = value.toString();
		//check proper format
		Matcher matcher = pattern.matcher(valStr);
		if (matcher.matches()){
			//extract stock symbol
			int substrBegin = valStr.indexOf(htmlGetStr);
			int substrEnd = valStr.indexOf(htmlExtStr);
			String stock = valStr.substring(substrBegin + htmlGetStr.length(), substrEnd);
			joinKey.set(stock);
			joinKeyPair.setFirst(joinKey);

			//extract the log
			int endOfLog = valStr.indexOf(quoteSymbol, substrEnd);
			joinVal.set(valStr.substring(0, endOfLog + 1));
			valuePair.setFirst(joinVal);
			context.write(joinKeyPair, valuePair);
		}
		else{
			log.warn("bad weblog input: " + value.toString());
			context.getCounter("Assig4", "Bad Web Log Entry").increment(1);
		}
	}
}
